//
// connection.hpp
// ~~~~~~~~~~~~~~
//
// Copyright (c) 2003-2013 Christopher M. Kohlhoff (chris at kohlhoff dot com)
//
// Distributed under the Boost Software License, Version 1.0. (See accompanying
// file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)
//

#ifndef SERIALIZATION_CONNECTION_HPP
#define SERIALIZATION_CONNECTION_HPP


/*
<<<<<<< .mine
=======


#include <boost/shared_ptr.hpp>
#include <boost/tuple/tuple.hpp>
#include <boost/asio/detail/atomic_count.hpp>
#include <iomanip>
#include <string>
#include <sstream>
#include <vector>
#include <exception>
#include <boost/asio/strand.hpp>

#include "logger.hpp"
#include "protocol.h"

>>>>>>> .r238
*/

#include <boost/archive/text_iarchive.hpp>
#include <boost/archive/text_oarchive.hpp>
#include <boost/tuple/tuple.hpp>
#include <boost/asio/detail/atomic_count.hpp>
#include <boost/foreach.hpp>

#include <boost/serialization/vector.hpp>

namespace rpc {

/// The connection class provides serialization primitives on top of a socket.
/**
* Each message sent using this class consists of:
* @li An 8-byte header containing the length of the serialized data in
* hexadecimal.
* @li The serialized data.
*/
class connection
{
public:
	/// Constructor.
	connection(boost::asio::io_service& io_service)
	: socket_(io_service)
	, strand_(io_service)
	{
		std::cerr << "connection()" << std::endl;
	}

	~connection() 
	{
		std::cerr << "~connection()" << std::endl;
	}

	/// Get the underlying socket. Used for making a connection or for accepting
	/// an incoming connection.
	boost::asio::ip::tcp::socket& socket()
	{
		return socket_;
	}

	// 同步写
	template<typename T>
	size_t write(T& t, int time_out = -1) {
		// Serialize the data first so we know how large it is.
		std::ostringstream archive_stream;
		boost::archive::text_oarchive archive(archive_stream);
		archive << t;
		outbound_data_ = archive_stream.str();

		// Format the header.
		out_pkg_header_.sign[0] = (char)'w';
		out_pkg_header_.sign[1] = (char)'y';
		out_pkg_header_.sign[2] = (char)'x';
		out_pkg_header_.sign[3] = (char)'x';
		out_pkg_header_.size = outbound_data_.size();		
		
		
		// Write the serialized data to the socket. We use "gather-write" to send
		// both the header and the data in a single write operation.
		std::vector<boost::asio::const_buffer> buffers;
		buffers.push_back(boost::asio::buffer((void*)&out_pkg_header_, sizeof(rpc::protocol::package_header)));
		buffers.push_back(boost::asio::buffer(outbound_data_));

		boost::system::error_code error;
		size_t bytes_write = boost::asio::write(socket_, buffers);

		return bytes_write;

		/*
		boost::system::error_code error;
		size_t bytes_write = boost::asio::write(socket_, boost::asio::buffer(buffer), error);
		if(error) {
			close();
			return -1;
		}

		return bytes_write;
		*/
	}

	// 同步读数据
	template<typename T>
	size_t read(T& t, int time_out = -1) {
		boost::system::error_code error;

		size_t bytes_read = boost::asio::read(socket_, 
			boost::asio::buffer((void*)&in_pkg_header_, sizeof(protocol::package_header)));
		
		// Determine the length of the serialized data.
		if(!rpc::protocol::is_valid_package(in_pkg_header_)) {
			// Header doesn't seem to be valid. Inform the caller.
			boost::system::error_code error(boost::asio::error::invalid_argument);
			return -1;
		}

		// Start an asynchronous call to receive the data.
		inbound_data_.resize(in_pkg_header_.size);
		bytes_read = boost::asio::read(socket_, boost::asio::buffer(inbound_data_));

		std::string archive_data(&inbound_data_[0], inbound_data_.size());
		std::istringstream archive_stream(archive_data);
		boost::archive::text_iarchive archive(archive_stream);
		archive >> t;

		return bytes_read;
	}

	/// Asynchronously write a data structure to the socket.
	template <typename T, typename Handler>
	void async_write(const T& t, Handler handler)
	{
		// Serialize the data first so we know how large it is.
		std::ostringstream archive_stream;
		boost::archive::text_oarchive archive(archive_stream);
		archive << t;
		outbound_data_ = archive_stream.str();

		// Format the header.
		out_pkg_header_.sign[0] = (char)'w';
		out_pkg_header_.sign[1] = (char)'y';
		out_pkg_header_.sign[2] = (char)'x';
		out_pkg_header_.sign[3] = (char)'x';
		out_pkg_header_.size = outbound_data_.size();		
		
		// Write the serialized data to the socket. We use "gather-write" to send
		// both the header and the data in a single write operation.
		std::vector<boost::asio::const_buffer> buffers;
		buffers.push_back(boost::asio::buffer((void*)&out_pkg_header_, sizeof(protocol::package_header)));
		buffers.push_back(boost::asio::buffer(outbound_data_));

		void (connection::*f)(const boost::system::error_code&, boost::tuple<Handler>)
			= &connection::handle_write_completed<Handler>;

		boost::asio::async_write(socket_, 
			buffers, 
			boost::bind(f, this,
			boost::asio::placeholders::error, 
			boost::make_tuple(handler)));
	}

	template<typename Handler>
	void handle_write_completed(const boost::system::error_code& e, boost::tuple<Handler> handler) {
	  boost::get<0>(handler)(e, this);
	}

	/// Asynchronously read a data structure from the socket.
	template <typename T, typename Handler>
	void async_read(T& t, Handler handler)
	{
		// Issue a read operation to read exactly the number of bytes in a header.
		void (connection::*f)(
			const boost::system::error_code&,
			T&, boost::tuple<Handler>)
			= &connection::handle_read_header<T, Handler>;

		boost::asio::async_read(socket_, 
			boost::asio::buffer((void*)&in_pkg_header_, sizeof(protocol::package_header)),
			boost::bind(f,
			this, boost::asio::placeholders::error, boost::ref(t),
			boost::make_tuple(handler)));
	}

	/// Handle a completed read of a message header. The handler is passed using
	/// a tuple since boost::bind seems to have trouble binding a function object
	/// created using boost::bind as a parameter.
	template <typename T, typename Handler>
	void handle_read_header(const boost::system::error_code& e,
		T& t, boost::tuple<Handler> handler)
	{
		if (e)
		{
			boost::get<0>(handler)(e, this);
		}
		else
		{
			// Determine the length of the serialized data.
			if(!rpc::protocol::is_valid_package(in_pkg_header_)) {
				// Header doesn't seem to be valid. Inform the caller.
				boost::system::error_code error(boost::asio::error::invalid_argument);
				boost::get<0>(handler)(error, this);
				return;
			}

			// Start an asynchronous call to receive the data.
			inbound_data_.resize(in_pkg_header_.size);
			void (connection::*f)(
				const boost::system::error_code&,
				T&, boost::tuple<Handler>)
				= &connection::handle_read_data<T, Handler>;
			boost::asio::async_read(socket_, boost::asio::buffer(inbound_data_),
				boost::bind(f, this,
				boost::asio::placeholders::error, boost::ref(t), handler));
		}
	}

	/// Handle a completed read of message data.
	template <typename T, typename Handler>
	void handle_read_data(const boost::system::error_code& e, T& t, boost::tuple<Handler> handler)
	{
		if (e)
		{
			boost::get<0>(handler)(e, this);
		}
		else
		{
			// Extract the data structure from the data just received.
			try
			{
				std::string archive_data(&inbound_data_[0], inbound_data_.size());
				std::istringstream archive_stream(archive_data);
				boost::archive::text_iarchive archive(archive_stream);
				archive >> t;
			}
			catch (std::exception& e)
			{
				// Unable to decode data.
				boost::system::error_code error(boost::asio::error::invalid_argument);
				boost::get<0>(handler)(error, this);
				return;
			}

			// Inform caller that data has been received ok.
			boost::get<0>(handler)(e, this);
		}
	}

	void close() 
	{
		socket_.close();
	}


	bool is_timeout() { return false; };  

// attributes
public:
	boost::uint32_t id() { return id_; }  
	boost::uint16_t get_business_type(){ return business_type_; }  
	void set_business_type(boost::uint16_t type) { business_type_ = type; }  
	boost::uint32_t get_app_id(){ return app_id_; }  
	void set_app_id(boost::uint32_t app_id) { app_id_ = app_id; }  
	std::string& get_remote_addr() { return name_; }  
	void set_remote_addr(const std::string& name) { name_ = name; }  

private:
	/// The underlying socket.
	boost::asio::ip::tcp::socket socket_;
	static boost::asio::detail::atomic_count last_id_;  

	boost::uint32_t id_;  
	boost::uint16_t  business_type_;  
	boost::uint32_t app_id_;  
	std::string name_;  
	boost::asio::strand strand_;  
	std::time_t last_op_time_;

	/// Holds the in/outbound data.
	rpc::protocol::package_header out_pkg_header_;
	rpc::protocol::package_header in_pkg_header_;
	std::string outbound_data_;
	std::vector<char> inbound_data_;
}; // class xconnection

typedef boost::shared_ptr<rpc::connection> connection_ptr;

} // namespace rpc

#endif // SERIALIZATION_CONNECTION_HPP

